package com.coupon.config;

import com.alibaba.fastjson.JSON;
import com.common.domain.dto.UserCouponAddDTO;
import com.coupon.service.UserCouponService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @author 路亚
 * @version 0.1
 * @className MQConfig
 * @description
 * @date 2024/12/23 15:16
 * @since jdk11
 */
@Configuration
public class MQConfig {

    @Value("${luya.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @Bean
    public DefaultMQProducer defaultMQProducer() {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("video");
        defaultMQProducer.setNamesrvAddr(namesrvAddr);
        try {
            defaultMQProducer.start();
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        return defaultMQProducer;
    }

    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer(UserCouponService userCouponService){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("video");
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            consumer.subscribe("video", "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt messageExt : list) {
                        UserCouponAddDTO userCouponAddDTO = JSON.parseObject(new String(messageExt.getBody()), UserCouponAddDTO.class);
                        userCouponService.addUserCoupon(userCouponAddDTO);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        return consumer;
    }
}
